package org.jianfa.mq.seckill.rocketmq;

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.jianfa.mq.seckill.entity.po.OrderInfo;
import org.jianfa.mq.seckill.redis.RedisService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

@Service
public class RocketMqProducer {

    private static final Logger logger = LoggerFactory.getLogger(RocketMqProducer.class);

    private DefaultMQProducer producer;

    @PostConstruct
    public void postConstrct() throws Exception{
        producer = new DefaultMQProducer("jianfa");
        producer.setNamesrvAddr("127.0.0.1:9876");
        producer.start();
    }

    @PreDestroy
    public void preDestroy() {
        producer.shutdown();
    }

    public boolean sendMessage(SeckillMessage seckillMessage){

        try {
            String message = RedisService.beanToString(seckillMessage);
            /*
             * Create a message instance, specifying topic, tag and message body.
             */
            Message msg = new Message("Seckill" /* Topic */,
                    message.getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
            );

            /*
             * Call send message to deliver message to one of brokers.
             */

            SendResult sendResult = producer.send(msg);

            SendStatus sendStatus = sendResult.getSendStatus();
            return SendStatus.SEND_OK.equals(sendStatus) ? true : false;
        } catch (Exception ex) {
            logger.error(String.format("发送消息失败。userId:%s goodsId:%s", seckillMessage.getUserId(), seckillMessage.getGoodsId()), ex);
            return false;
        }
    }

    public boolean sendDelayMessage(OrderInfo orderInfo) {
        try {
            String message = RedisService.beanToString(orderInfo);
            /*
             * Create a message instance, specifying topic, tag and message body.
             */
            Message msg = new Message("OrderInfo" /* Topic */,
                    message.getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
            );

            /*
             * 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
             */
            msg.setDelayTimeLevel(6);

            /*
             * Call send message to deliver message to one of brokers.
             */

            SendResult sendResult = producer.send(msg);

            SendStatus sendStatus = sendResult.getSendStatus();
            return SendStatus.SEND_OK.equals(sendStatus) ? true : false;
        } catch (Exception ex) {
            logger.error(String.format("发送消息失败。orderId:{} userId:{} goodsId:{}", orderInfo.getId(), orderInfo.getUserId(), orderInfo.getGoodsId()), ex);
            return false;
        }
    }
}
